package com.example.lt.service.impl;

import com.example.lt.constant.RedisConstant;
import com.example.lt.context.BaseContext;
import com.example.lt.entity.user.Model;
import com.example.lt.entity.user.User;
import com.example.lt.entity.user.UserModel;
import com.example.lt.entity.video.Video;
import com.example.lt.service.InterestPushService;
import com.example.lt.service.video.TypeService;
import com.example.lt.util.RedisUtil;
import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @Author 小涛Tao
 * @Date: 2024/08/04/21:04
 * version 1.0 注释：
 **/
@Service
@RequiredArgsConstructor
public class InterestPushServiceImpl implements InterestPushService {

    private final TypeService typeService;

    private final RedisUtil redisUtil;

    private final RedisTemplate redisTemplate;

    /**
     * 推入标签库，把视频 id 添加到所在的标签库中
     * @param video
     */
    @Override
    @Async
    public void pushSystemStockIn(Video video) {
        // 往系统的标签库中添加视频id
        final List<String> labels = video.buildLabel();
        final Long videoId = video.getId();
        // 使用通道的方式进行添加
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sAdd((RedisConstant.SYSTEM_STOCK + label).getBytes(), String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    /**
     * 添加分类库，把视频 id 添加到所在的分类库中，用于后续随机推送分类视频
     * @param video
     */
    @Override
    @Async
    public void pushSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        redisUtil.sSet(RedisConstant.SYSTEM_TYPE_STOCK + typeId, video.getId());
    }

    /**
     * 根据分类获取视频id，乱序随机
     * @param typeId
     * @return
     */
    @Override
    public Collection<Long> listVideoIdByTypeId(Long typeId) {
        // 随机获取12个视频的id
        final List<Object> ids = redisTemplate.opsForSet().randomMembers(RedisConstant.SYSTEM_TYPE_STOCK + typeId, 12);
        // 因为 list 中可能会有 null，所以进行去 null 和 去重
        final HashSet<Long> videoIds = new HashSet<>();
        for (Object id : ids) {
            videoIds.add(Long.parseLong(id.toString()));
        }
        return videoIds;
    }

    /**
     * 删除标签内视频
     * @param video
     */
    @Override
    @Async
    public void deleteSystemStockIn(Video video) {
        final List<String> labels = video.buildLabel();
        final Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sRem((RedisConstant.SYSTEM_STOCK + label).getBytes(),String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    /**
     * 删除分类库中的视频
     * @param video
     */
    @Override
    @Async
    public void deleteSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        redisUtil.setRemove(RedisConstant.SYSTEM_TYPE_STOCK + typeId, video.getId());
    }

    /**
     * 用户初始化模型 -> 订阅分类
     * @param userId 用户id
     * @param labels 标签名
     * @param oldLabels 旧标签
     */
    @Override
    @Async
    public void initUserModel(Long userId, List<String> labels, List<String> oldLabels) {
        final String key = RedisConstant.USER_MODEL + userId;

        // 创建 Map 集合存储标签及其分数
        Map<Object, Object> modelMap = new HashMap<>();
        if (!ObjectUtils.isEmpty(labels)) {
            final int size = labels.size();
            // 将标签分为等分概率,不可能超过100个分类
            double probabilityValue = 100 / size;
            for (String label : labels) {
                modelMap.put(label, probabilityValue);
            }
        }
        // 删除redis中旧的标签
        redisUtil.hdel(key, oldLabels);
        // 添加标签到redis中
        redisUtil.hmset(key, modelMap);
    }

    /**
     * 修改用户模型
     * @param userModel 用户模型相关信息
     */
    @Override
    @Async
    public void updateUserModel(UserModel userModel) {
        final Long userId = BaseContext.get();
        // 如果【userId】为null，则说明不是登录状态，不用执行修改模型的代码
        if (userId != null) {
            final List<Model> models = userModel.getModels();
            // 获取当前用户的模型
            final String key = RedisConstant.USER_MODEL + userId;
            Map<Object, Object> modelMap = redisUtil.hmget(key);

            if (modelMap == null) {
                modelMap = new HashMap<>();
            }

            // 修改用户模型
            for (Model model : models) { // 遍历每一个【标签】+【分数】的封装对象
                final String strKey = model.getLabel();
                if (modelMap.containsKey(strKey)) { // 如果【存在】该标签了，则在原来的基础上加减【分数】
                    modelMap.put(strKey, Double.parseDouble(modelMap.get(strKey).toString()) + model.getScore());
                } else {
                    modelMap.put(strKey, model.getScore());
                }
                Object o = modelMap.get(strKey);
                if (o == null || Double.parseDouble(o.toString()) < 0.0) {
                    modelMap.remove(strKey);
                }
            }

            // 每个标签概率同等加上标签数，再同等除以标签数  防止数据膨胀
            final int labelSize = modelMap.keySet().size();
            for (Object o : modelMap.keySet()) {
                modelMap.put(o, (Double.parseDouble(modelMap.get(o).toString()) + labelSize) / labelSize);
            }

            redisUtil.hmset(key, modelMap); // 更新用户模型
        }
    }

    /**
     * 根据标签获取视频id【随机】
     * @param labelNames 标签集合
     * @return
     */
    @Override
    public Object listVideoIdByLabels(ArrayList<String> labelNames) {
        final ArrayList<String> labelKeys = new ArrayList<>();
        for (String labelName : labelNames) {
            labelKeys.add(RedisConstant.SYSTEM_STOCK + labelName);
        }
        Set<Long> videoIds = new HashSet<>();
        final List<Object> list = redisUtil.sRandom(labelKeys);
        if (!ObjectUtils.isEmpty(list)) {
            videoIds = list.stream()
                    .filter(id -> !ObjectUtils.isEmpty(id))
                    .map(id -> Long.valueOf(id.toString()))
                    .collect(Collectors.toSet());
        }
        return videoIds;
    }

    /**
     * 用户获取推送视频【兴趣推送】，其中有一条视频是和性别有关
     * @param user 用户信息
     * @return 视频id集合
     */
    @Override
    public Collection<Long> listVideoIdByUserModel(User user) {
        // 创建结果集，一次最多推送10个视频
        Set<Long> videoIds = new HashSet<>(10);

        if (!ObjectUtils.isEmpty(user)) {
            final Long userId = user.getId();
            // 从用户模型中拿命中率
            final Map<Object, Object> modelMap = redisUtil.hmget(RedisConstant.USER_MODEL + userId);
            if (!ObjectUtils.isEmpty(modelMap)) {
                // 根据用户模型获取对应的标签概率数组
                final String[] probabilityArray =  initProbabilityArray(modelMap);
                // 随机选择【probabilityArray】中的标签，后续一个标签就要去redis中随机获取一个【视频id】
                final Random randomNum = new Random();
                final ArrayList<String> labelNames = new ArrayList<>(); // 用于保存从【probabilityArray】中随机选择的标签
                // 随机获取标签
                for (int i=0; i<8; i++) {
                    String labelName = probabilityArray[randomNum.nextInt(probabilityArray.length)];
                    labelNames.add(labelName);
                }
                // 提升性能
                String str1 = RedisConstant.SYSTEM_STOCK;
                // 随机获取
                List<Object> list = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (String labelName : labelNames) {
                        String key = str1 + labelName;
                        connection.sRandMember(key.getBytes());
                    }
                    return null;
                });
                // 获取视频id集合【videoIds】
                Set<Long> ids = list.stream()
                        .filter(id -> !ObjectUtils.isEmpty(id))
                        .map(id -> Long.parseLong(id.toString()))
                        .collect(Collectors.toSet());

                String str2 = RedisConstant.HOT_VIDEO;
                // 进行去重，查询这些视频id的历史记录
                List simpIds = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (Long id : ids) {
                        connection.get((str2 + id + ":" + userId).getBytes());
                    }
                    return null;
                });
                simpIds = (List) simpIds.stream().filter(o -> !ObjectUtils.isEmpty(o)).collect(Collectors.toList());
                if (!ObjectUtils.isEmpty(simpIds)) {
                    for (Object simpId : simpIds) {
                        final Long l = Long.valueOf(simpId.toString());
                        if (ids.contains(l)) {
                            ids.remove(l);
                        }
                    }
                }

                videoIds.addAll(ids);

                final Boolean sex = user.getSex();
                // 随机挑选一个视频,根据性别: 男：美女 女：宠物
                final Long oneId = randomVideoId(sex);
                if (oneId != null) {
                    videoIds.add(oneId);
                }

                return videoIds; // 返回结果
            }
        }

        // 如果【user为空】的话，就说明是游客，那么就随机的来推送10个视频
        // 获取10个标签
        final List<String> labels = typeService.random10Labels();
        final ArrayList<String> labelKey = new ArrayList<>();
        final int size = labels.size();
        final Random random = new Random();
        // 随机获取标签
        for (int i=0; i<size; i++) {
            final int randomIndex = random.nextInt(size);
            labelKey.add(RedisConstant.SYSTEM_STOCK + labels.get(randomIndex));
        }
        // 获取视频id
        final List<Object> videoIdList = redisUtil.sRandom(labelKey);
        if (!ObjectUtils.isEmpty(videoIdList)) {
            // 过滤掉空的元素，并将其转换成long
            videoIds = videoIdList.stream()
                    .filter(id -> !ObjectUtils.isEmpty(id))
                    .map(id -> Long.valueOf(id.toString()))
                    .collect(Collectors.toSet());
        }
        // 返回视频id集合
        return videoIds;
    }

    /**
     * 根据性别随机获取一个视频id
     * @param sex 性别
     * @return
     */
    public Long randomVideoId(Boolean sex) {
        String key = RedisConstant.SYSTEM_STOCK + (sex ? "美女" : "宠物");
        final Object o = redisUtil.sRandom(key);
        if (!ObjectUtils.isEmpty(o)) {
            return Long.parseLong(o.toString());
        }
        return null;
    }

    /**
     * 初始化概率数组 -> 保存的元素是标签
     * @param modelMap 用户模型，【key】是标签，【value】是分数
     * @return
     */
    private String[] initProbabilityArray(Map<Object, Object> modelMap) {
        // key: 标签  value：概率
        Map<String, Integer> probabilityMap = new HashMap<>();
        final int size = modelMap.size();
        final AtomicInteger n = new AtomicInteger(0);
        modelMap.forEach((k, v) -> {
            // 这里防止【probability】为0，或者是【probability】过大，就同等加上标签数再除标签数
            int probability = (((Double)v).intValue() + size) / size;
            n.addAndGet(probability);
            probabilityMap.put(k.toString(), probability);
        });
        final String[] probabilityArray = new String[n.get()];
        final AtomicInteger index = new AtomicInteger(0); // 索引
        // 初始化数组
        probabilityMap.forEach((label, p) -> {
            int i = index.get();
            int limit = i + p;
            while(i < limit) {
                probabilityArray[i++] = label;
            }
            index.set(limit);
        });
        return probabilityArray;
    }
}
